package com.learn.datasource.custom;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.UUID;

/**
 * @create: 2023-04-18 10:05
 * @author: Mr.Du
 * --------------
 * @notes: 基于RichParallelSourceFunction创建并行并带有rich功能的Source
 **/
public class CustomRichSourceWithParallelDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> in = env.addSource(new MyRichParallelSource());
        in.print();

        env.execute();
    }

}

/**
 * Rich类型的Source比非Rich的多出：
 * open方法，实例化的时候会执行一次，多个并行度会执行多次（因为是多个实例了）
 * close方法，销毁实例的时候会执行一次，多个并行度会执行多次
 * getRuntime方法可以获得当前的Runtime对象（底层API）
 */
class MyRichParallelSource extends RichParallelSourceFunction<String> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        System.out.println("open.....");
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        ctx.collect(UUID.randomUUID().toString());
    }

    @Override
    public void close() throws Exception {
        super.close();
        System.out.println("close....");
    }

    @Override
    public void cancel() {
        System.out.println("cancel...");
    }


}
